package com.demo.kafka.producer;

import com.alibaba.fastjson.JSON;
import com.demo.kafka.consumer.KafkaConsumer;
import com.demo.kafka.entity.KafkaMessage;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.TopicPartition;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Set;

/**
 * 测试Kafka发送消息
 * @author cyj
 */
@Slf4j
@Service
public class KafkaProducer {

    @Resource
    private KafkaTemplate<String, KafkaMessage> kafkaTemplate;

    @Resource
    private KafkaConsumer kafkaConsumer;

    /**
     * 发送消息
     * @param kafkaMessage 消息
     */
    @SneakyThrows
    public void send(KafkaMessage kafkaMessage) {
        Object object = kafkaTemplate.send("test-topic-1", kafkaMessage).get();
        log.info("result：{}", JSON.toJSONString(object));
    }

    /**
     * 发送消息并指定分配
     * @param kafkaMessage 消息
     */
    @SneakyThrows
    public void sendPartition0(KafkaMessage kafkaMessage) {
        Object object = kafkaTemplate.send("test-topic-1", 0, kafkaMessage.getMsgId(), kafkaMessage).get();
        log.info("sendPartition0 result：{}", JSON.toJSONString(object));
    }

    /**
     * 发送消息并指定分配
     * @param kafkaMessage 消息
     */
    @SneakyThrows
    public void sendPartition1(KafkaMessage kafkaMessage) {
        Object object = kafkaTemplate.send("test-topic-1", 1, kafkaMessage.getMsgId(), kafkaMessage).get();
        log.info("sendPartition1 result：{}", JSON.toJSONString(object));
    }

    /**
     * 发送消息，consumer获取消息
     * @param kafkaMessage 消息
     * @param partition 分区
     * @return 分区信息
     */
    @SneakyThrows
    public Set<TopicPartition> sendAndGetMessage(KafkaMessage kafkaMessage, int partition) {
        Object object = kafkaTemplate.send("test-topic-1", partition, kafkaMessage.getMsgId(), kafkaMessage).get();
        log.info("sendAndGetPartition result：{}", JSON.toJSONString(object));

        return kafkaConsumer.consumer();
    }

    /**
     * 发送消息到分区1
     * @param kafkaMessage 消息
     * @return 分区信息
     */
    @SneakyThrows
    public Set<TopicPartition> sendAndGetMessagePartition1(KafkaMessage kafkaMessage) {
        Object object = kafkaTemplate.send("test-topic-1", 1, kafkaMessage.getMsgId(), kafkaMessage).get();
        log.info("sendAndGetPartition result：{}", JSON.toJSONString(object));

        return kafkaConsumer.consumerPartition1();
    }

    /**
     * 发送消息到指定分区，两个consumer是否收到消息
     * @param kafkaMessage 消息
     * @param partition 分区
     */
    @SneakyThrows
    public void sendAndGetMessageForPartition(KafkaMessage kafkaMessage, int partition) {
        Object object = kafkaTemplate.send("test-topic-1", partition, kafkaMessage.getMsgId(), kafkaMessage).get();
        log.info("sendAndGetPartition result：{}", JSON.toJSONString(object));

        kafkaConsumer.consumer();
        kafkaConsumer.consumerPartition1();
    }
}
